package com.sunzm.spark.streaming.callcenter;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Date;
import java.util.Properties;

/**
 * 生成呼叫中心kafka的模拟数据
 */
public class CallCenterMockData {
    public static void main(String[] args) {
        String bootstrapServers = "82.156.210.70:9093";
        //服务器上运行，使用内网
        //String bootstrapServers = "10.0.8.11:9092";
        String topic = "call-center-message";

        Properties properties = new Properties();

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        //指定key和value使用的序列化方式，消费者需要反序列化，才能得到原始的消息内容
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //创建kafka生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        try {

            //呼叫方向： 0呼入； 1呼出
            int direction = 0;
            //int direction = 1;
            if (direction == 0) {
                //模拟呼入的数据
                //公司ID
                String companyId = "cmp1001";
                //消息类型 start通话开始; staff-ring坐席振铃; staff-answer坐席接听; staff-hunup坐席挂机;
                // user-ring：用户振铃； user-answer：用户接听; user-hunup：用户挂断; stop:通话结束
                String messageType = "start";
                //坐席ID
                String staffId = "stf1001";
                //坐席名称
                String staffName = "客服1001";
                //获取当前时间
                long currentTimeMillis = System.currentTimeMillis();
                //时间 yyyyMMddHHmmssSSS
                String dateTime = DateFormatUtils.format(currentTimeMillis, "yyyyMMddHHmmssSSS");
                //通话ID
                String callId = StringUtils.substring(dateTime, 0, 12) + "001";

                //构造消息
                //通话开始
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("companyId", companyId);
                jsonObject.put("messageType", messageType);
                jsonObject.put("staffId", staffId);
                jsonObject.put("staffName", staffName);
                jsonObject.put("callId", callId);
                jsonObject.put("dateTime", dateTime);

                producer.send(new ProducerRecord<String, String>(topic, callId, jsonObject.toJSONString()));

                //坐席振铃
                messageType = "staff-ring";
                jsonObject.put("messageType", messageType);
                //获取振铃时间: +5秒
                Date ringDate = DateUtils.addSeconds(DateUtils.parseDate(dateTime, "yyyyMMddHHmmssSSS"), 5);
                dateTime = DateFormatUtils.format(ringDate, "yyyyMMddHHmmssSSS");
                jsonObject.put("dateTime", dateTime);
                producer.send(new ProducerRecord<String, String>(topic, callId, jsonObject.toJSONString()));

                //坐席接听
                messageType = "staff-answer";
                jsonObject.put("messageType", messageType);
                //获取接听时间: 当前时间+10秒
                Date answerDate = DateUtils.addSeconds(DateUtils.parseDate(dateTime, "yyyyMMddHHmmssSSS"), 10);
                dateTime = DateFormatUtils.format(answerDate, "yyyyMMddHHmmssSSS");
                jsonObject.put("dateTime", dateTime);
                producer.send(new ProducerRecord<String, String>(topic, callId, jsonObject.toJSONString()));

                //坐席挂断，假设通话时长2分钟
                messageType = "staff-hunup";
                jsonObject.put("messageType", messageType);
                //获取挂断时间: 当前时间+120秒
                Date hunupDate = DateUtils.addSeconds(DateUtils.parseDate(dateTime, "yyyyMMddHHmmssSSS"), 120);
                dateTime = DateFormatUtils.format(hunupDate, "yyyyMMddHHmmssSSS");
                jsonObject.put("dateTime", dateTime);
                producer.send(new ProducerRecord<String, String>(topic, callId, jsonObject.toJSONString()));

                //通话结束
                messageType = "stop";
                jsonObject.put("messageType", messageType);
                //获取结束时间: 当前时间+2秒
                Date endDate = DateUtils.addSeconds(DateUtils.parseDate(dateTime, "yyyyMMddHHmmssSSS"), 2);
                dateTime = DateFormatUtils.format(endDate, "yyyyMMddHHmmssSSS");
                jsonObject.put("dateTime", dateTime);
                producer.send(new ProducerRecord<String, String>(topic, callId, jsonObject.toJSONString()));
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭
            producer.close();
        }
    }
}
